package net.bwie.realtime.jtp.job;

import net.bwie.realtime.jtp.common.utils.KafkaUtil;
import net.bwie.realtime.jtp.function.CdcDataEtlMapFunction;
import net.bwie.realtime.jtp.utils.MysqlCdcUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 简介说明:
 *
 * @author: LiLi
 * @date: 2025/06/02 14:12:27
 * @version: 1.0
 */
public class JtpCdcMallDataJob {
    public static void main(String[] args) throws Exception {
        // 1. 执行环境-env

        // 1.1  启用检查点（Checkpoint）并设置检查点间隔为 5000 毫秒（即 5 秒
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        env.enableCheckpointing(5000L);

        // 2.数据源-source

//        DataStream<String> cdcStream = MysqlCdcUtil.cdcMysqlRaw(env, "jtp_mall", "order_info");

//        DataStream<String> cdcStream = MysqlCdcUtil.cdcMysqlDeser(env, "jtp_mall", "order_info");

        DataStream<String> cdcStream = cdcMysqlSource(env);

//        cdcStream.print();
        SingleOutputStreamOperator<String> etlStream = cdcStream.map(new CdcDataEtlMapFunction());
//        etlStream.print();

        // 3.输出到kfk主题
         KafkaUtil.producerKafka(etlStream, "topic-jtp-mall");

        env.execute("JtpCdcMallDataJob");
    }

    private static DataStream<String> cdcMysqlSource(StreamExecutionEnvironment env) {
//        todo:为维度表添加数据源 全量 + 增量

        String[] dimList = new String[]{
                "base_dic", "base_province", "base_region", "base_category1", "base_category2", "base_category3",
                "activity_info", "activity_rule", "base_trademark", "coupon_info", "sku_info", "spu_info", "user_info"
        };

        DataStream<String> cdcDimStream = MysqlCdcUtil.cdcMysqlInitial(env, "jtp_mall", dimList);


//      todo: 为事实表添加数据源  增量
        String[] factList = new String[]{
                "favor_info", "cart_info", "order_info", "order_detail", "order_detail_activity", "order_detail_coupon",
                "order_status_log", "payment_info", "order_refund_info", "refund_payment", "comment_info", "coupon_use"
        };

        DataStream<String> cdcFactStream = MysqlCdcUtil.cdcMysqlEarliest(env, "jtp_mall", factList);

//        todo: 合并 数据流

        DataStream<String> cdcStream = cdcDimStream.union(cdcFactStream);

//        cdcStream.print();

        return cdcStream;
    }
}
